package acs.controller;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import acs.cat.Cat;
import acs.cat.message.Transaction;
import acs.collection.Params;
import acs.constant.EventType;

public class MessageQueue {
   private static Logger logger = LoggerFactory.getLogger(MessageQueue.class);
   private static final long WARN_COST = 2;
   private MessageObserver observer;
   private static MessageQueue ourInstance=new MessageQueue();
   private LinkedBlockingQueue<MessageItem> messageQueue = new LinkedBlockingQueue<>();
   //store the maps which are not in used
   private ConcurrentLinkedQueue<MessageItem> itemPool = new ConcurrentLinkedQueue<>();
   class MessageItem {

    int eventType;
    Params param;
   }
   
   public static MessageQueue getInstance() {
    return ourInstance;
   }

   
   public void eventLoop() {
	Transaction transaction = null;
    while (true) {
      try {
        MessageItem item = messageQueue.take();
        transaction = Cat.newTransaction("event", EventType.getDescription(item.eventType));
        long startTime = System.currentTimeMillis();
        observer.handle(item.eventType, item.param);
        long cost = System.currentTimeMillis() - startTime;
        if (cost > WARN_COST) {
          logger.warn("handle event {} cost long time: {} ms", item.eventType, cost);
        }
        item.param = null;
        itemPool.add(item);
      } catch (Exception e) {
        logger.error("Failed in event loop ! cause: {}", e);
      }finally {
      }
    }
   }
   
   public void putMessage(int eventType, Params param) {
	MessageItem item = retrieveItem();
    item.eventType = eventType;
    item.param = param;
    try {
      messageQueue.put(item);
    } catch (Exception e) {
      logger.error("Failed when put msg!", e);
    }
  }
  private MessageItem retrieveItem() {
    MessageItem item = itemPool.poll();
    if (null == item) {
      item = new MessageItem();
    }
    return item;
  }
  
  public void registerObserver(MessageObserver observer) {
    this.observer = observer;
  }
}
